package org.caijin.springzookeeper.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributorLock implements Lock {
    // zk Server
    public  static String cluster_connection = "192.168.88.100:2181,192.168.88.101:2181,192.168.88.102:2181";
    private  int sessionTimeout = 5000;
    private ZooKeeper zk;
    private String rootNode = "locks";
    private String subNode = "seq-";
    // 当前client创建的子节点
    private String currentNode;
    // 当前client等待的子节点
    private String waitPath;
    private CountDownLatch connectDownLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    public DistributorLock()  {
        try {
            zk = new ZooKeeper(cluster_connection, sessionTimeout, new Watcher() {

                @Override
                public void process(WatchedEvent watchedEvent) {
                    // 连接建立时, 打开latch, 唤醒wait在该latch上的线程
                    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                        connectDownLatch.countDown();
                    }
                    // 发生了waitPath的删除事件
                    if (watchedEvent.getType() == Event.EventType.NodeDeleted&&watchedEvent.getPath().equals(waitPath)) {
                        waitLatch.countDown();
                    }
                }

            });
            // 等待连接建立
            connectDownLatch.await();
            //获取根节点状态
            Stat stat = zk.exists("/" + rootNode, false);

            //如果根节点不存在，则创建根节点，根节点类型为永久节点
            if (stat == null) {
                System.out.println("根节点不存在");
                zk.create("/"+rootNode,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

        }catch (Exception e) {
            e.printStackTrace();
        };
    }

    @Override
    public void lock() {
        try{
            //在根节点下创建临时顺序节点，返回值为创建的节点路径
            currentNode = zk.create("/"+rootNode+"/"+subNode,null,ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // wait一小会, 让结果更清晰一些
            Thread.sleep(50);
            // 注意, 没有必要监听"/locks"的子节点的变化情况
            List<String> childrenNodes = zk.getChildren("/" + rootNode, false);

            // 列表中只有一个子节点, 那肯定就是currentNode , 说明client获得锁
            if (childrenNodes.size() == 1) {
                return;
            }else {
                //对根节点下的所有临时顺序节点进行从小到大排序
                Collections.sort(childrenNodes);
                //当前节点名称
                String thisNode = currentNode.substring(("/"+rootNode+"/").length());
                //获取当前节点的位置
                int index = childrenNodes.indexOf(thisNode);
                if (index == -1) {
                    System.out.println("数据异常");
                } else if (index == 0) {
                    // index == 0, 说明thisNode在列表中最小, 当前client获得锁
                    return;
                } else {
                    // 获得排名比currentNode 前1位的节点
                    this.waitPath = "/" + rootNode + "/" + childrenNodes.get(index - 1);
                    // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
                    zk.getData(waitPath, true, new Stat());
                    //进入等待锁状态
                    waitLatch.await();
                    return;
                }
            }

        }catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public void unlock() {
        try {
            zk.delete(this.currentNode,-1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
